package com.deep.flink.kafka;

import com.alibaba.fastjson.JSON;
import com.google.common.base.Strings;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;

import java.util.Properties;

public class KafkaExample {
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "192.168.1.85:9092,192.168.1.86:9092,192.168.1.87:9092");
        properties.setProperty("group.id", "g2");
        DeserializationSchema<String> deserializationSchema = new SimpleStringSchema();
        String topic = "customer_statusChangedEvent";
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<String> text = env.addSource(
                new FlinkKafkaConsumer010<String>(topic, deserializationSchema, properties));

        DataStream<Tuple2<String, Integer>> dataStream = text
                .filter(p -> !Strings.isNullOrEmpty(p))
                .map(new MapFunction<String, Tuple2<String, Integer>>() {
                    @Override
                    public Tuple2<String, Integer> map(String p) throws Exception {
                        CustomerStatusChangedEvent event = JSON.parseObject(p, CustomerStatusChangedEvent.class);
                        return new Tuple2<String, Integer>(Long.toString(event.getCustomerId()), event.getNewStatus());
                    }
                });

        dataStream.print();
        try {
            env.execute("Flink-Kafka");
        } catch (Exception ex) {

        }
    }
}

